package com.hitqz.robot.driver.zmq.protocol.component;

import com.hitqz.robot.driver.zmq.protocol.message.ProtocolMessage;
import lombok.extern.slf4j.Slf4j;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;

/**
 * 基于ZeroMQ发送消息线程
 */
@Slf4j
public class SendThread extends Thread {

    private final String backName;

    public SendThread(String backName) {
        this.backName = backName;
    }

    @Override
    public void run() {
        try (ZContext ctx = new ZContext()) {
            configureContext(ctx);
            ZMQ.Socket worker = configureSocket(ctx);
            boolean connect = worker.connect("ipc://" + backName);
            log.info("[wheel][zmq]sendThread connect {} result:{}", backName, connect);

            while (true) {
                Thread.sleep(1);
                if (!MsgQueue.isEmpty()) {
                    ProtocolMessage protocolMessage = MsgQueue.deleteOne();
                    if (protocolMessage != null) {
                        sendProtocolMessage(worker, protocolMessage);
                    }
                }
            }
        } catch (Exception e) {
            log.error("[wheel][zmq]sendThread start error:", e);
            restartSendThread();
        }
    }

    private void configureContext(ZContext ctx) {
        ctx.setRcvHWM(0);
        ctx.setSndHWM(0);
        ctx.setLinger(0);
    }

    private ZMQ.Socket configureSocket(ZContext ctx) {
        ZMQ.Socket worker = ctx.createSocket(SocketType.DEALER);
        worker.setRcvHWM(0);
        worker.setSndHWM(0);
        worker.setLinger(0);
        worker.setReconnectIVL(500);
        worker.setSendTimeOut(1000);

        worker.setTCPKeepAlive(1);
        worker.setTCPKeepAliveCount(3);
        worker.setTCPKeepAliveIdle(60L);
        worker.setTCPKeepAliveInterval(10L);

        worker.setSendBufferSize(1024 * 10);
        worker.setReceiveBufferSize(1024 * 10);

        return worker;
    }

    private void sendProtocolMessage(ZMQ.Socket worker, ProtocolMessage protocolMessage) {
        worker.sendMore(protocolMessage.getAddress());
        worker.sendMore(protocolMessage.getType());
        boolean isSuccess = worker.send(protocolMessage.getData());
        if (!isSuccess) {
            log.error("[wheel][zmq]failed to send message [{}], retry",protocolMessage.getAddress());
            retrySendMessage(worker,protocolMessage);
        }
    }

    private void retrySendMessage(ZMQ.Socket worker, ProtocolMessage protocolMessage) {
        int maxRetries = 3; // 最大重试次数
        int retryCount = 0;

        while (retryCount < maxRetries) {
            try {
                Thread.sleep(1000); // 每次重试前等待1秒
                worker.sendMore(protocolMessage.getAddress());
                worker.sendMore(protocolMessage.getType());
                boolean isSuccess = worker.send(protocolMessage.getData());
                if (isSuccess) {
                    log.info("[wheel][zmq]message [{}] sent successfully after retry",protocolMessage.getAddress());
                    return;
                }
                retryCount++;
            } catch (InterruptedException e) {
                log.error("[wheel][zmq]sendThread interrupted during retry", e);
                Thread.currentThread().interrupt(); // 重新设置中断标志
            }
        }
        log.error("[wheel][zmq]failed to send message [{}] after {} retries",protocolMessage.getAddress(), maxRetries);
    }


    private void restartSendThread() {
        Thread sendThread = new Thread(new SendThread(backName));
        sendThread.start();
        log.warn("[wheel][zmq]restart sendThread");
    }
}
